package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Demo3MySource {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //使用自定义source
        DataStream<Integer> myDS = env.addSource(new MySource());

        myDS.print();

        env.execute();


    }
}

/**
 * 自定义source
 * 实现SourceFunction接口，实现接口中的run方法
 */
class MySource implements SourceFunction<Integer> {
    /**
     * flink启动的时候会执行一次，再run方法中读取外部的数据，将数据发送到下游
     */
    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {

        while (true){
            //将数据发送到下游
            ctx.collect(100);

            Thread.sleep(100);
        }
    }

    //cancel方法再任务取消的时候执行，一般用于回收资源
    @Override
    public void cancel() {
    }
}
